package com.realtime.temp;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.realtime.util.ClickHouseUtil;
import com.realtime.util.JsonDebeziumDeserializationUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Properties;

import static org.apache.flink.table.api.Expressions.$;

public class DwsCouponReduceRate {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //每家店铺求每天优惠券的补贴率
        SourceFunction<String> mySqlSource = MySQLSource.<String>builder()
                .hostname("hadoop101")
                .port(3306)
                .databaseList("gmall") // set captured database
                .tableList("gmall.order_info_realtime,gmall.store") // set captured table
                .username("root")
                .password("root")
                .debeziumProperties((Properties) new Properties().setProperty("LOG_LEVEL","ERROR"))
                .deserializer(new JsonDebeziumDeserializationUtil()) // converts SourceRecord to JSON String
                .build();
        DataStreamSource<String> streamSource = env.addSource(mySqlSource);
        OutputTag<String> store_stream = new OutputTag<String>("store"){};


        SingleOutputStreamOperator<String> process = streamSource.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String tb = jsonObject.getString("table");
                String data = jsonObject.getString("data");
                if (tb.equals("store")) {
                    ctx.output(store_stream, data);
                } else {
                    out.collect(data);
                }
            }
        });
        process.print();
        DataStream<String> store = process.getSideOutput(store_stream);
        SingleOutputStreamOperator<OrderInfos> orderInfoStream = process.map(new MapFunction<String, OrderInfos>() {

            @Override
            public OrderInfos map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String orderId = jsonObject.getString("id");
                String status = jsonObject.getString("order_status");
                Double totalAmount = jsonObject.getDouble("total_amount");
                Double couponReduceAmount = jsonObject.getDouble("coupon_reduce_amount");
                Long ts = jsonObject.getLong("create_time");
                String store_id = jsonObject.getString("store_id");
                OrderInfos orderInfo = new OrderInfos(orderId, status, totalAmount,couponReduceAmount, ts, store_id);
                return orderInfo;
            }
        }).assignTimestampsAndWatermarks( WatermarkStrategy
                .<OrderInfos>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                .withTimestampAssigner((event, timestamp) -> event.getTs())).filter(new FilterFunction<OrderInfos>() {
            @Override
            public boolean filter(OrderInfos value) throws Exception {
                return value.getStatus().equals("1002");
            }
        });

        SingleOutputStreamOperator<Store> storeBeanStream = store.map(new MapFunction<String, Store>() {
            @Override
            public Store map(String value) throws Exception {
                return JSONObject.parseObject(value, Store.class);
            }
        });
        tEnv.createTemporaryView("order_info", orderInfoStream,$("orderId"),$("status"),$("totalAmount"),$("store_id"),$("couponReduceAmount"),$("times").rowtime());
        tEnv.createTemporaryView("store", storeBeanStream);
        Table table = tEnv.sqlQuery("select window_start,window_end,store_id,storeName,sum_total_amount,sum_coupon_reduce_amount,UNIX_TIMESTAMP() as `times` from (SELECT window_start, window_end,store_id,SUM(totalAmount) sum_total_amount,sum(couponReduceAmount) sum_coupon_reduce_amount\n" +
                "  FROM TABLE(\n" +
                "    TUMBLE(TABLE order_info, DESCRIPTOR(times), INTERVAL '60' MINUTES))\n" +
                "  GROUP BY window_start, window_end,store_id) t1 left join store on store.storeID=t1.store_id");
        DataStream<Tuple2<Boolean, ResultBeanReduceRate>> tuple2DataStream = tEnv.toRetractStream(table, ResultBeanReduceRate.class);

        SingleOutputStreamOperator<ResultBeanReduceRate> result = tuple2DataStream.map(new MapFunction<Tuple2<Boolean, ResultBeanReduceRate>, ResultBeanReduceRate>() {
            @Override
            public ResultBeanReduceRate map(Tuple2<Boolean, ResultBeanReduceRate> value) throws Exception {
                return value.f1;
            }
        });
        SinkFunction<ResultBeanReduceRate> jdbcSink = ClickHouseUtil.<ResultBeanReduceRate>getJdbcSink("insert into result_bean_reduce_rate values(?,?,?,?,?,?,?)");
        result.addSink(jdbcSink);

        env.execute();

/*
drop table if exists result_bean_reduce_rate;
create table if not exists result_bean_reduce_rate
(
    stt           DateTime,
    edt           DateTime,
    store_id   String,
    store_name String,
    sum_order_amount  Decimal(16, 2),
    sum_coupon_reduce_amount  Decimal(16, 2),
    ts            UInt64
) engine = ReplacingMergeTree(ts)
      partition by toYYYYMMDD(stt)
      order by (stt, edt, store_id);
SELECT * FROM result_bean_reduce_rate;

select store_name,dates,concat(toString((total_reduce_amount/total_order_amount)*100),'%') reduce_rate
from (select store_name, toYYYYMMDD(stt) dates, sum(sum_coupon_reduce_amount) total_reduce_amount , sum(sum_order_amount) total_order_amount
      from result_bean_reduce_rate
      where toYYYYMMDD(stt) = toYYYYMMDD(now())
      GROUP BY store_name, toYYYYMMDD(stt)) sntYss;

*/

    }
}

